package output

import (
	"encoding/json"
	"errors"
	"github.com/Shopify/sarama"
	"time"
)

type Option struct {
	Retry   int
	Address []string
}

type KafkaClient struct {
	kafka  sarama.SyncProducer		// 同步
	option *Option
}

type KafkaMessage struct {
	Data      map[string]interface{}
	Timestamp int64
}

func NewKafka(option *Option) *KafkaClient {
	config := sarama.NewConfig()
	config.Producer.Compression = sarama.CompressionSnappy  // message zip
	config.Producer.Flush.Frequency = 10 * time.Millisecond // flush batches every 10ms
	config.Producer.Flush.MaxMessages = 1 << 29
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Return.Successes = true
	config.Version = sarama.V0_10_0_0 // kafka version
	config.Producer.Retry.Max = 3
	config.Producer.Idempotent = true	// 设置消息幂等性
	if option.Retry > 0 {
		config.Producer.Retry.Max = option.Retry
	}
	config.Producer.Partitioner = sarama.NewHashPartitioner // hash partition
	producer, err := sarama.NewSyncProducer(option.Address, config)  // sarama.Logger = zaplogger
	if err != nil {
		panic(err)
	}
	return &KafkaClient{
		kafka:  producer,
		option: option,
	}
}

// Send Message
func (k *KafkaClient) Publish(datas ...*WalData) error {
	if len(datas) == 0 {
		return errors.New("data is empty")
	}
	var messages []*sarama.ProducerMessage
	for _, data := range datas {
		key, topic, message := k.handleData(data)
		msgByte, err := json.Marshal(message)
		if err != nil {
			continue
		}
		messages = append(messages, &sarama.ProducerMessage{
			Topic: topic,
			Key:   sarama.StringEncoder(key),
			Value: sarama.ByteEncoder(msgByte),
		})
	}
	if len(messages) == 0{
		return errors.New("message is empty")
	}
	return k.kafka.SendMessages(messages)
}

// Handle Message Data
func (k *KafkaClient) handleData(data *WalData) (key, topic string, message *KafkaMessage) {
	return data.Key, data.Topic, &KafkaMessage{
		Data:      data.Data,
		Timestamp: data.Timestamp,
	}
}

// Close Kafka
func (k *KafkaClient) Close() {
	defer k.kafka.Close()
}


/*
	Kafka发送消息主要有三种方式：

	1.发送并忘记 2.同步发送 3.异步发送+回调函数

	方式一：发送并忘记(不关心消息是否正常到达，对返回结果不做任何判断处理)

	方式二：同步发送(通过get方法等待Kafka的响应，判断消息是否发送成功)

	方式三：异步发送+回调函数(消息以异步的方式发送，通过回调函数返回消息发送成功/失败)


三种方式虽然在时间上有所差别，但并不是说时间越快的越好，具体要看业务的应用场景：

场景1：如果业务要求消息必须是按顺序发送的，那么可以使用同步的方式，并且只能在一个partation上，结合参数设置retries的值让发送失败时重试，设置max_in_flight_requests_per_connection=1，可以控制生产者在收到服务器晌应之前只能发送1个消息，从而控制消息顺序发送；

场景2：如果业务只关心消息的吞吐量，容许少量消息发送失败，也不关注消息的发送顺序，那么可以使用发送并忘记的方式，并配合参数acks=0，这样生产者不需要等待服务器的响应，以网络能支持的最大速度发送消息；

场景3：如果业务需要知道消息发送是否成功，并且对消息的顺序不关心，那么可以用异步+回调的方式来发送消息，配合参数retries=0，并将发送失败的消息记录到日志文件中；


// https://github.com/JeffreyDing11223/kago/blob/master/syncProducer.go
 */